package com.atguigu.chapter11.function;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/7/26 9:11
 */
public class Flink02_Function_Table {
    public static void main(String[] args) {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        Table table = tEnv.fromValues("hello atguigu", "atguigu hello", "abc a b c");
        
        // 1. table
        // 内联
        /*table
//            .joinLateral(call(SplitAndLen.class, $("f0")))  // 内连接
            .leftOuterJoinLateral(call(SplitAndLen.class, $("f0")))  // 内连接
            .select($("f0"), $("w"), $("len"))
            .execute()
            .print();*/
        // 注册
       /* tEnv.createTemporaryFunction("split", SplitAndLen.class);
        table
//                        .joinLateral(call("split", $("f0")))  // 内连接
            .leftOuterJoinLateral(call(SplitAndLen.class, $("f0")))  // 内连接
            .select($("f0"), $("w"), $("len"))
            .execute()
            .print();*/
        
        
        // 2. sql
        tEnv.createTemporaryView("t1", table);
        
        tEnv.createTemporaryFunction("split", SplitAndLen.class);
        
        // 内连接的写法1:
        /*tEnv.sqlQuery("select " +
                          " f0," +
                          " w," +
                          " len " +
                          "from t1 " +
                          "join lateral table(split(f0)) on true")
            .execute()
            .print();*/
        
        // select ... from a,b where a.id=b.id; ====  select .. from a join b on a.id=b.id
        tEnv.sqlQuery("select " +
                          " f0," +
                          " w1," +
                          " len1 " +
                          "from t1, " +
                          " lateral table(split(f0)) as t(w1,len1)")
            .execute()
            .print();
        
    }
    
    @FunctionHint(output = @DataTypeHint("row<w string, len int>"))
    public static class SplitAndLen extends TableFunction<Row> {
        public void eval(String s) {
            if (s.contains("hello")) return;
            String[] words = s.split(" ");
            
            for (String word : words) {
                // 调用几次就生成几行
                collect(Row.of(word, word.length()));
            }
        }
    }
    
}
/*
hello  5
atguigu  6

 */
